package cas.ihep.fs.alluxio;

import alluxio.client.block.AlluxioBlockStore;
import alluxio.client.block.BlockWorkerInfo;
import alluxio.client.block.stream.BlockInStream;
import alluxio.client.file.FileSystemContext;
import alluxio.client.file.options.InStreamOptions;
import alluxio.client.file.options.OutStreamOptions;
import alluxio.proto.dataserver.Protocol;
import alluxio.wire.WorkerNetAddress;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Random;

class AlluxioInvoker {

    private final AlluxioBlockStore mStore;
    private final Method ref_createNettyBlockInStream;
    private final Random mRandom;

    AlluxioInvoker(FileSystemContext mCtx)throws Exception{
        mStore=AlluxioBlockStore.create(mCtx);
        ref_createNettyBlockInStream=BlockInStream.class.getDeclaredMethod(
                "createNettyBlockInStream", FileSystemContext.class,WorkerNetAddress.class,
                BlockInStream.BlockInStreamSource.class,Protocol.ReadRequest.class,
                long.class, InStreamOptions.class);
        ref_createNettyBlockInStream.setAccessible(true);
        Field random=AlluxioBlockStore.class.getDeclaredField("mRandom");
        random.setAccessible(true);
        mRandom=(Random) random.get(mStore);
    }

    final int nextInt(int bound){
        return mRandom.nextInt(bound);
    }

    final InputStream createNettyBlockInStream(FileSystemContext ctx, WorkerNetAddress addr,
                                               BlockInStream.BlockInStreamSource source, Protocol.ReadRequest req,
                                               long val, InStreamOptions opts) throws InvocationTargetException, IllegalAccessException {
        return (InputStream)ref_createNettyBlockInStream.invoke(null,ctx,addr,source,req,val,opts);
    }

    final List<BlockWorkerInfo> getWorkerInfoList() throws IOException {
        return mStore.getWorkerInfoList();
    }

    final OutputStream getOutStream(long id, long size, WorkerNetAddress addr, OutStreamOptions opts)throws IOException{
        return mStore.getOutStream(id, size, addr, opts);
    }
}
